package io.renren.kafka.consume;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Arrays;
import java.util.Date;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.renren.htby.utils.KafkaConsumeUtils;
import io.renren.modules.iots.entity.EqmStateEntity;
import io.renren.modules.iots.entity.EquipmentStateHistoryEntity;
import io.renren.modules.iots.entity.IotsEquipmentInfoEntity;
import io.renren.modules.iots.service.EquipmentStateHistoryService;
import io.renren.modules.iots.service.IotsEquipmentInfoService;
import io.renren.runner.EqmStateConsumerRunner;

/**
 * 设备状态监控 的Consumer
 * 
 * @author lfy.xys
 * @date 2018年5月31日
 *
 */
@Component
public class EqmStateConsumer {

	private KafkaConsumer<String, String> kafkaConsumer = null;

	private static Logger logger = LoggerFactory.getLogger(EqmStateConsumer.class);

	private static DateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

	@Autowired
	private IotsEquipmentInfoService iotsEquipmentInfoService;
	@Autowired
	private EquipmentStateHistoryService equipmentStateHistoryService;

	public void setConsumer() {
		Properties props = new Properties();
		// bootstrap.servers 和Producer一样，是指向Kafka集群的IP地址，以逗号分隔。
		props.put("bootstrap.servers", KafkaConsumeUtils.bootstrapServers);
		props.put("group.id", KafkaConsumeUtils.eqm_state_group_id);// 使用 监控的组
		props.put("enable.auto.commit", KafkaConsumeUtils.enable_auto_commit);
		props.put("auto.commit.interval.ms", KafkaConsumeUtils.auto_commit_interval_ms);
		props.put("auto.offset.reset", KafkaConsumeUtils.auto_offset_reset);
		props.put("session.timeout.ms", KafkaConsumeUtils.session_timeout_ms);
		// 发序列化。Consumer把来自Kafka集群的二进制消息反序列化为指定的类型。因本例中的Producer使用的是String类型，所以调用StringDeserializer来反序列化
		props.put("key.deserializer", KafkaConsumeUtils.key_deserializer);
		props.put("value.deserializer", KafkaConsumeUtils.value_deserializer);
		kafkaConsumer = new KafkaConsumer<String, String>(props);
	}

	public void consume(String topic) {
		if (kafkaConsumer == null) {
			this.setConsumer();
		}
		// Consumer订阅了Topic的消息，
		kafkaConsumer.subscribe(Arrays.asList(topic));
		logger.info("订阅了topic:{}", topic);
		while (true) {
			// Consumer调用poll方法来轮循Kafka集群的消息，其中的参数100是超时时间（Consumer等待直到Kafka集群中没有消息为止）：
			ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
			for (ConsumerRecord<String, String> record : records) {
				String msg = record.value();
				// logger.info("--------------- offset ={}", record.offset());
				updateState(msg);
			}
		}
	}
	
	/**
	 * 更新设备信息
	 * @author lfy.xys
	 * @date 2018年6月21日
	 *
	 * @param msg
	 */
	public void updateState(String msg) {
		// 设备状态： http://domain/device_status/IOServer标识/分组标识/设备标识/status_lastonlinetime
		String[] msgs = msg.split("/");
		Long key = Long.parseLong(msgs[msgs.length - 2]);
		logger.info("设备标识是 {}", key);

		String[] strs = msgs[msgs.length - 1].split("_");

		// 获取 map 中的 设备
		EqmStateEntity entity = EqmStateConsumerRunner.eqmMap.get(key);
		if (entity != null) {
			// 存在设备，判断并 更新

			Integer oldState = entity.getState();
			Integer newState = Integer.parseInt(strs[0]);
			if (oldState != newState) {
				// 状态改变，需要入库
				IotsEquipmentInfoEntity eqmEntity = iotsEquipmentInfoService.selectById(key);
				if(eqmEntity == null) {
					logger.info("没有查询到此设备id : {}，需要手动同步设备入库。", key);
					return;
				}
				
				eqmEntity.setState(newState);
				eqmEntity.setUpdatetime(new Date());
				iotsEquipmentInfoService.updateById(eqmEntity);

				// 插入历史记录
				EquipmentStateHistoryEntity eqmStateEntity = new EquipmentStateHistoryEntity();
				eqmStateEntity.setEqmid(key);
				eqmStateEntity.setState(newState);
				equipmentStateHistoryService.insert(eqmStateEntity);
			}

		} else {
			// 不存在，新加
			entity = new EqmStateEntity();
			
		}
		
		entity.setId(key);
		entity.setState(Integer.parseInt(strs[0]));
		try {
			entity.setUpdatetime(df.parse(strs[1]));
		} catch (ParseException e) {
			e.printStackTrace();
		}
		
		// 更新 内存 中的数据
		EqmStateConsumerRunner.eqmMap.put(key, entity);
	}
	
	/**
	 * 测试消费 的业务
	 * @author lfy.xys
	 * @date 2018年6月21日
	 *
	 */
	public void TestConsume() {
		int i = 0;
		while(i++ < 100) {
			String msg = "http://domain/device_status/IOServer标识/分组标识/1/" + i%2 + "_2018-06-13 13:34:42";
			updateState(msg);
			
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

	public static void main(String[] args) {
		String[] a = "http://domain/device_status/IOServer标识/分组标识/设备标识/status_lastonlinetime".split("/");
		System.out.println(a.length);
		for (String string : a) {
			System.out.println(string);
		}
		System.out.println(a[a.length - 2]);
	}
}
